package com.chengqs.rapid.console;

import com.chengqs.rapid.console.consumer.MQConsumerFactory;
import com.chengqs.rapid.console.consumer.MetricConsumer;
import lombok.extern.slf4j.Slf4j;

import java.util.Map;

/**
 * <B>主类名称：</B>ConsumerContainer<BR>
 * <B>概要说明：</B>enter<BR>
 *
 * @author: chengqs
 * @create: 2025-03-18 17:51:33
 **/
@Slf4j
public class ConsumerContainer {
    public ConsumerContainer(RapidConsoleProperties rapidConsoleProperties) {
        for(int i = 0; i < rapidConsoleProperties.getConsumerNum(); i ++) {
            MQConsumerFactory.getInstance().createConsumer(rapidConsoleProperties.getKafkaAddress(),
                    rapidConsoleProperties.getGroupId(),
                    rapidConsoleProperties.getTopicNamePrefix(),
                    i);
        }
    }

    public void start() {
        runReactorConsumer();
    }

    public void stop() {
        MQConsumerFactory.getInstance().stopConsumers();
    }

    private void runReactorConsumer() {
        for(Map.Entry<String, MetricConsumer> me : MQConsumerFactory.getConsumers().entrySet()) {
            MetricConsumer mqConsumer = me.getValue();
            mqConsumer.start();
        }
        log.info("ConsumerServer started...");
    }
}
